package com.dailyblue.java.spring.rocketmq.顺序消息;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerOrderly;
import org.apache.rocketmq.common.message.MessageExt;

import java.io.UnsupportedEncodingException;
import java.util.List;

public class 顺序消息的消费 {
    public static void main(String[] args) throws Exception{
        // 创建消费者 Consumer，指定消费者组名
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("zhuoyue02");
        // 指定 NameSrv 地址
        consumer.setNamesrvAddr("192.168.160.88:9876");
        // 订阅主题 Topic 和 Tag
        consumer.subscribe("TopicC", "*");
        consumer.registerMessageListener(new MessageListenerOrderly() {
            @Override
            public ConsumeOrderlyStatus consumeMessage(List<MessageExt> list, ConsumeOrderlyContext consumeOrderlyContext) {
                for(MessageExt me:list){
                    try {
                        String name = Thread.currentThread().getName();
                        String body = new String(me.getBody(),"utf-8");
                        System.out.println(name+":"+body);
                    } catch (UnsupportedEncodingException e) {
                        return ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT;
                    }
                }
                return ConsumeOrderlyStatus.SUCCESS;
            }
        });
        consumer.start();
    }
}
